package Thread;


import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 模拟生产者线程
 * **/
class ProdecerThread extends Thread{

    private BlockingQueue<String> queue;  //队列
    //volatile关键字 保证线程之间可见
    private volatile Boolean flag = true;  //让线程停止的标志
    private static AtomicInteger count = new AtomicInteger();  //并发包下原子类，保证多个线程之间可见

    public ProdecerThread(BlockingQueue queue){
        this.queue = queue;
    }

    public void stopThread(){
        this.flag = false;
    }

    @Override
    public void run() {
        try{
            System.out.println("生产者线程启动++++++");
            while(this.flag){
               System.out.println("正在生产数据:");
               String data = count.incrementAndGet()+"";//执行+1操作
               Boolean success = queue.offer(data,2, TimeUnit.SECONDS);
               if(success){
                   System.out.println("生产者，生产数据成功:"+data);
               }else{
                   System.out.println("生产者，生产数据失败:"+data);
               }
               Thread.sleep(1000);
            }
        }catch (Exception e){
           e.printStackTrace();
        }finally {
            System.out.println("生产者退出线程");
        }
    }
}

/**
 * 模拟消费者线程
 * **/
class ConsumerThread extends Thread{

    private BlockingQueue<String> queue;  //队列
    private volatile Boolean flag = true;

    public ConsumerThread(BlockingQueue<String> blockingQueue){
        this.queue = blockingQueue;
    }


    @Override
    public void run() {
        try{
            System.out.println("消费线程启动...");
            while(this.flag){
                String data =  queue.poll(2,TimeUnit.SECONDS);
                System.out.println("消费者消费线程成功...."+data);
                if(data!=null){
                    System.out.println("消费成功:"+data);
                }else{
                    System.out.println("消费失败");
                    this.flag = false;
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            System.out.println("消费者退出线程");
        }
    }
}

/***
 * 使用BlockingQueue模拟生产者与消费者
 * */
public class ThreadProducerConsumer {

    public static void main(String[] args) throws Exception{

        BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);
        ProdecerThread p1 = new ProdecerThread(queue);
        //ProdecerThread p2 = new ProdecerThread(queue);
        ConsumerThread c = new ConsumerThread(queue);
        p1.start();
        //p2.start();
        c.start();

        Thread.sleep(10*1000);
        p1.stopThread();
        //p2.stopThread();
    }
}
